/* Copyright  2009
   @Author
   Richard Changde Yin            e-mail yinchangde@hotmail.com

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

#include "heapdef.h"

#include <common.h>
#include <error.h>

#include <decimal.h>

#include <string.h>

static int keys_compare(heap_rb_param *param, unsigned char *key1, unsigned char *key2);
static void init_block(HP_BLOCK *block,uint32_t chunk_length, uint32_t min_records,
                        uint32_t max_records);

#define FIXED_REC_OVERHEAD (sizeof(unsigned char))
#define VARIABLE_REC_OVERHEAD (sizeof(unsigned char**) + ALIGN_SIZE(sizeof(unsigned char)))

/* Minimum size that a chunk can take, 12 bytes on 32bit, 24 bytes on 64bit */
#define VARIABLE_MIN_CHUNK_SIZE \
        ((sizeof(unsigned char**) + VARIABLE_REC_OVERHEAD + sizeof(unsigned char**) - 1) & ~(sizeof(unsigned char**) - 1))


/* Create a heap table */

int heap_create(const char *name, uint32_t keys, HP_KEYDEF *keydef,
    uint32_t columns, HP_COLUMNDEF *columndef,
    uint32_t max_key_fieldnr, uint32_t key_part_size,
    uint32_t reclength, uint32_t keys_memory_size,
    uint32_t max_records, uint32_t min_records,
    HP_CREATE_INFO *create_info, HP_SHARE **res)
{
  uint32_t i, j, key_segs, max_length, length;
  uint32_t max_rows_for_stated_memory;
  HP_SHARE *share= 0;
  HA_KEYSEG *keyseg;

  if (!create_info->internal_table)
  {
    pthread_mutex_lock(&THR_LOCK_heap);
    if ((share= hp_find_named_heap(name)) && share->open_count == 0)
    {
      hp_free(share);
      share= 0;
    }
  }

  if (!share)
  {
    size_t chunk_dataspace_length;
    uint32_t chunk_length, is_variable_size;
    uint32_t fixed_data_length, fixed_column_count;
    HP_KEYDEF *keyinfo;

    if (create_info->max_chunk_size)
    {
      uint32_t configured_chunk_size= create_info->max_chunk_size;

      /* User requested variable-size records, let's see if they're possible */

      if (configured_chunk_size < key_part_size)
      {
        /* Eventual chunk_size cannot be smaller than key data,
           which allows all keys to fit into the first chunk */
        my_error(ER_CANT_USE_OPTION_HERE, MYF(0), "block_size");
        pthread_mutex_unlock(&THR_LOCK_heap);
        return(ER_CANT_USE_OPTION_HERE);
      }

      if ((reclength - configured_chunk_size) >= VARIABLE_MIN_CHUNK_SIZE<<1)
      {
        /* Allow variable size only if we're saving at least two smallest chunks */
        /* There has to be at least one field after indexed fields */
        /* Note that NULL bits are already included in key_part_size */
        is_variable_size= 1;
        chunk_dataspace_length= configured_chunk_size;
      }
      else
      {
        /* max_chunk_size is near the full reclength, let's use fixed size */
        is_variable_size= 0;
        chunk_dataspace_length= reclength;
      }
    }
    else if (create_info->is_dynamic)
    {
      /* User asked for dynamic records - use 256 as the chunk size */
      if ((key_part_size + VARIABLE_REC_OVERHEAD) > 256)
        chunk_dataspace_length= key_part_size;
      else
        chunk_dataspace_length= 256 - VARIABLE_REC_OVERHEAD;

      is_variable_size= 1;
    }
    else
    {
      /* if max_chunk_size is not specified, put the whole record in one chunk */
      is_variable_size= 0;
      chunk_dataspace_length= reclength;
    }

    if (is_variable_size)
    {
      /* Check whether we have any variable size records past key data */
      uint32_t has_variable_fields= 0;

      fixed_data_length= key_part_size;
      fixed_column_count= max_key_fieldnr;

      for (i= max_key_fieldnr; i < columns; i++)
      {
        HP_COLUMNDEF* column= columndef + i;
        if (column->type == COLLAPSAR_TYPE_VARCHAR && column->length >= 32)
        {
            /* The field has to be >= 5.0.3 true VARCHAR and have substantial length */
            /* TODO: do we want to calculate minimum length? */
            has_variable_fields= 1;
            break;
        }

        if (has_variable_fields)
        {
          break;
        }

        if ((column->offset + column->length) <= chunk_dataspace_length)
        {
          /* Still no variable-size columns, add one fixed-length */
          fixed_column_count= i + 1;
          fixed_data_length= column->offset + column->length;
        }
      }

      if (!has_variable_fields)
      {
        /* There is no need to use variable-size records without variable-size columns */
        /* Reset sizes if it's not variable size anymore */
        is_variable_size= 0;
        chunk_dataspace_length= reclength;
        fixed_data_length= reclength;
        fixed_column_count= columns;
      }
    }
    else
    {
      fixed_data_length= reclength;
      fixed_column_count= columns;
    }

    /*
      We store unsigned char* del_link inside the data area of deleted records,
      so the data length should be at least sizeof(unsigned char*)
    */
    set_if_bigger(chunk_dataspace_length, sizeof (unsigned char**));

    if (is_variable_size)
    {
      chunk_length= chunk_dataspace_length + VARIABLE_REC_OVERHEAD;
    }
    else
    {
      chunk_length= chunk_dataspace_length + FIXED_REC_OVERHEAD;
    }

    /* Align chunk length to the next pointer */
    chunk_length= (uint) (chunk_length + sizeof(unsigned char**) - 1) & ~(sizeof(unsigned char**) - 1);



    for (i= key_segs= max_length= 0, keyinfo= keydef; i < keys; i++, keyinfo++)
    {
      memset(&keyinfo->block, 0, sizeof(keyinfo->block));
      memset(&keyinfo->rb_tree , 0, sizeof(keyinfo->rb_tree));
      for (j= length= 0; j < keyinfo->keysegs; j++)
      {
         length+= keyinfo->seg[j].length;
         if (keyinfo->seg[j].null_bit)
        {
          length++;
          if (!(keyinfo->flag & HA_NULL_ARE_EQUAL))
                keyinfo->flag|= HA_NULL_PART_KEY;
          if (keyinfo->algorithm == HA_KEY_ALG_BTREE)
                keyinfo->rb_tree.size_of_element++;
      }
       switch (keyinfo->seg[j].type) {
        case HA_KEYTYPE_SHORT_INT:
        case HA_KEYTYPE_LONG_INT:
        case HA_KEYTYPE_FLOAT:
        case HA_KEYTYPE_DOUBLE:
        case HA_KEYTYPE_USHORT_INT:
        case HA_KEYTYPE_ULONG_INT:
        case HA_KEYTYPE_LONGLONG:
        case HA_KEYTYPE_ULONGLONG:
        case HA_KEYTYPE_INT24:
        case HA_KEYTYPE_UINT24:
        case HA_KEYTYPE_INT8:
         keyinfo->seg[j].flag|= HA_SWAP_KEY;
          break;
        case HA_KEYTYPE_VARBINARY1:
          /* Case-insensitiveness is handled in coll->hash_sort */
          keyinfo->seg[j].type= HA_KEYTYPE_VARTEXT1;
          /* fall_through */
        case HA_KEYTYPE_VARTEXT1:
          keyinfo->flag|= HA_VAR_LENGTH_KEY;
          length+= 2;
          /* Save number of bytes used to store length */
          keyinfo->seg[j].bit_start= 1;
          break;
        case HA_KEYTYPE_VARBINARY2:
          /* Case-insensitiveness is handled in coll->hash_sort */
          /* fall_through */
        case HA_KEYTYPE_VARTEXT2:
          keyinfo->flag|= HA_VAR_LENGTH_KEY;
          length+= 2;
          /* Save number of bytes used to store length */
          keyinfo->seg[j].bit_start= 2;
          /*
            Make future comparison simpler by only having to check for
            one type
          */
          keyinfo->seg[j].type= HA_KEYTYPE_VARTEXT1;
          break;
        default:
         break;
        }
      }
      keyinfo->length= length;
      length+= keyinfo->rb_tree.size_of_element +
                ((keyinfo->algorithm == HA_KEY_ALG_BTREE) ? sizeof(unsigned char*) : 0);
      if (length > max_length)
         max_length= length;
      key_segs+= keyinfo->keysegs;
      if (keyinfo->algorithm == HA_KEY_ALG_BTREE)
      {
        key_segs++; /* additional HA_KEYTYPE_END segment */
        if (keyinfo->flag & HA_VAR_LENGTH_KEY)
          keyinfo->get_key_length= hp_rb_var_key_length;
        else if (keyinfo->flag & HA_NULL_PART_KEY)
          keyinfo->get_key_length= hp_rb_null_key_length;
        else
          keyinfo->get_key_length= hp_rb_key_length;
      }
    }
    share= NULL;
    if (!(share= (HP_SHARE*) malloc(sizeof(HP_SHARE))))
      goto err;

    memset(share, 0, sizeof(HP_SHARE));

    if (keys && !(share->keydef= (HP_KEYDEF*) malloc(keys*sizeof(HP_KEYDEF))))
      goto err;

    memset(share->keydef, 0, keys*sizeof(HP_KEYDEF));

    if (keys && !(share->keydef->seg= (HA_KEYSEG*) malloc(key_segs*sizeof(HA_KEYSEG))))
      goto err;
    if (!(share->column_defs= (HP_COLUMNDEF*)
	  malloc(columns*sizeof(HP_COLUMNDEF))))
      goto err;

    memset(share->column_defs, 0, columns*sizeof(HP_COLUMNDEF));

    /*
       Max_records is used for estimating block sizes and for enforcement.
       Calculate the very maximum number of rows (if everything was one chunk) and
       then take either that value or configured max_records (pick smallest one)
    */
    max_rows_for_stated_memory= (ha_rows) (create_info->max_table_size /
      (keys_memory_size + chunk_length));
    max_records = ((max_records && max_records < max_rows_for_stated_memory) ?
                      max_records : max_rows_for_stated_memory);

    memcpy(share->column_defs, columndef, (size_t) (sizeof(columndef[0]) * columns));

    share->key_stat_version= 1;
    keyseg= keys ? share->keydef->seg : NULL;

    init_block(&share->recordspace.block, chunk_length, min_records, max_records);
    /* Fix keys */
    memcpy(share->keydef, keydef, (size_t) (sizeof(keydef[0]) * keys));
    for (i= 0, keyinfo= share->keydef; i < keys; i++, keyinfo++)
    {
      keyinfo->seg= keyseg;
      memcpy(keyseg, keydef[i].seg,
	     (size_t) (sizeof(keyseg[0]) * keydef[i].keysegs));
      keyseg+= keydef[i].keysegs;

      if (keydef[i].algorithm == HA_KEY_ALG_BTREE)
      {
        /* additional HA_KEYTYPE_END keyseg */
       keyseg->type=     HA_KEYTYPE_END;
       keyseg->length=   sizeof(unsigned char*);
       keyseg->flag=     0;
       keyseg->null_bit= 0;
       keyseg++;

       init_tree(&keyinfo->rb_tree, 0, 0, sizeof(unsigned char*),
                 (qsort_cmp2)keys_compare, 1, NULL, NULL);
              keyinfo->delete_key= hp_rb_delete_key;
              keyinfo->write_key= hp_rb_write_key;
      }
      else
      {
           init_block(&keyinfo->block, sizeof(HASH_INFO), min_records,
                       max_records);
        keyinfo->delete_key= hp_delete_key;
        keyinfo->write_key= hp_write_key;
        keyinfo->hash_buckets= 0;
      }
      if ((keyinfo->flag & HA_AUTO_KEY) && create_info->with_auto_increment)
        share->auto_key= i + 1;
    }
    share->min_records= min_records;
    share->max_records= max_records;
    share->max_table_size= create_info->max_table_size;
    share->index_length= 0;
    share->blength= 1;
    share->keys= keys;
    share->max_key_length= max_length;
    share->column_count= columns;
    share->changed= 0;
    share->auto_key= create_info->auto_key;
    share->auto_key_type= create_info->auto_key_type;
    share->auto_increment= create_info->auto_increment;

    share->fixed_data_length= fixed_data_length;
    share->fixed_column_count= fixed_column_count;

    share->recordspace.chunk_length= chunk_length;
    share->recordspace.chunk_dataspace_length= chunk_dataspace_length;
    share->recordspace.is_variable_size= is_variable_size;
    share->recordspace.total_data_length= 0;

    if (is_variable_size) {
      share->recordspace.offset_link= chunk_dataspace_length;
      share->recordspace.offset_status= share->recordspace.offset_link + sizeof(unsigned char**);
    } else {
      share->recordspace.offset_link= 1<<22; /* Make it likely to fail if anyone uses this offset */
      share->recordspace.offset_status= chunk_dataspace_length;
    }

    /* Must be allocated separately for rename to work */
    if (!(share->name= strdup(name)))
    {
      goto err;
    }
    thr_lock_init(&share->lock);
    pthread_mutex_init(&share->intern_lock,MY_MUTEX_INIT_FAST);
    if (!create_info->internal_table)
    {
      share->open_list.data= (void*) share;
      heap_share_list= list_add(heap_share_list,&share->open_list);
    }
    else
      share->delete_on_close= 1;
  }
  if (!create_info->internal_table)
    pthread_mutex_unlock(&THR_LOCK_heap);

  *res= share;
  return(0);

err:
  if(share && share->keydef && share->keydef->seg)
    free(share->keydef->seg);
  if(share && share->keydef)
    free(share->keydef);
  if(share && share->column_defs)
    free(share->column_defs);
  if(share)
    free(share);
  if (!create_info->internal_table)
    pthread_mutex_unlock(&THR_LOCK_heap);
  return(1);
} /* heap_create */


static int keys_compare(heap_rb_param *param, unsigned char *key1, unsigned char *key2)
{
  uint32_t not_used[2];
  return ha_key_cmp(param->keyseg, key1, key2, param->key_length,
                    param->search_flag, not_used);
}

static void init_block(HP_BLOCK *block, uint32_t chunk_length, uint32_t min_records,
                      uint32_t max_records)
{
  uint32_t i,recbuffer,records_in_block;

  max_records= cmax(min_records,max_records);
  if (!max_records)
    max_records= 1000;             /* As good as quess as anything */

  /* we want to start each chunk at 8 bytes boundary, round recbuffer to the next 8 */
  recbuffer= (uint) (chunk_length + sizeof(unsigned char**) - 1) & ~(sizeof(unsigned char**) - 1);
  records_in_block= max_records / 10;
  if (records_in_block < 10 && max_records)
    records_in_block= 10;
  if (!records_in_block || records_in_block*recbuffer >
      (default_record_cache_size-sizeof(HP_PTRS)*HP_MAX_LEVELS))
       records_in_block= (default_record_cache_size - sizeof(HP_PTRS) *
                     HP_MAX_LEVELS) / recbuffer + 1;
  block->records_in_block= records_in_block;
  block->recbuffer= recbuffer;
  block->last_allocated= 0L;

  for (i= 0; i <= HP_MAX_LEVELS; i++)
    block->level_info[i].records_under_level=
      (!i ? 1 : i == 1 ? records_in_block :
       HP_PTRS_IN_NOD * block->level_info[i - 1].records_under_level);
}


static inline void heap_try_free(HP_SHARE *share)
{
  if (share->open_count == 0)
    hp_free(share);
  else
    share->delete_on_close= 1;
}


int heap_delete_table(const char *name)
{
  int result;
  register HP_SHARE *share;

  pthread_mutex_lock(&THR_LOCK_heap);
  if ((share= hp_find_named_heap(name)))
  {
    heap_try_free(share);
    result= 0;
  }
  else
  {
    result= my_errno=ENOENT;
  }
  pthread_mutex_unlock(&THR_LOCK_heap);
  return(result);
}


void heap_drop_table(HP_INFO *info)
{
  pthread_mutex_lock(&THR_LOCK_heap);
  heap_try_free(info->s);
  pthread_mutex_unlock(&THR_LOCK_heap);
  return;
}


void hp_free(HP_SHARE *share)
{
  if (share->open_list.data)                    /* If not internal table */
    heap_share_list= list_delete(heap_share_list, &share->open_list);
  hp_clear(share);                               /* Remove blocks from memory */
  thr_lock_delete(&share->lock);
  pthread_mutex_destroy(&share->intern_lock);
  if(share->keys)
    free(share->keydef->seg);
  if(share->keys)
    free(share->keydef);
  free(share->column_defs);
  free((unsigned char*) share->name);
  free((unsigned char*) share);
  return;
}


/**
  Swap the contents of two variables.
 */
#define swap_variables(TYPE, a, b) \
  do {                             \
    TYPE dummy;                    \
    dummy= a;                      \
    a= b;                          \
    b= dummy;                      \
  } while (0)

#define CMP_NUM(a,b) (((a) < (b)) ? -1 : ((a) == (b)) ? 0 : 1)


int ha_compare_text(const CHARSET_INFO * const charset_info, unsigned char *a, uint32_t a_length,
                    unsigned char *b, uint32_t b_length, bool part_key,
                    bool skip_end_space)
{
  if (!part_key)
    return charset_info->coll->strnncollsp(charset_info, a, a_length,
                                           b, b_length, (bool)!skip_end_space);
  return charset_info->coll->strnncoll(charset_info, a, a_length,
                                       b, b_length, part_key);
}


static int compare_bin(unsigned char *a, uint32_t a_length, unsigned char *b, uint32_t b_length,
                       bool part_key, bool skip_end_space)
{
  uint32_t length= cmin(a_length,b_length);
  unsigned char *end= a+ length;
  int flag;

  while (a < end)
    if ((flag= (int) *a++ - (int) *b++))
      return flag;
  if (part_key && b_length < a_length)
    return 0;
  if (skip_end_space && a_length != b_length)
  {
    int swap= 1;
    /*
      We are using space compression. We have to check if longer key
      has next character < ' ', in which case it's less than the shorter
      key that has an implicite space afterwards.

      This code is identical to the one in
      strings/ctype-simple.c:my_strnncollsp_simple
    */
    if (a_length < b_length)
    {
      /* put shorter key in a */
      a_length= b_length;
      a= b;
      swap= -1; /* swap sign of result */
    }
    for (end= a + a_length-length; a < end ; a++)
    {
      if (*a != ' ')
	return (*a < ' ') ? -swap : swap;
    }
    return 0;
  }
  return (int) (a_length-b_length);
}


/*
  Compare two keys

  SYNOPSIS
    ha_key_cmp()
    keyseg  Array of key segments of key to compare
    a       First key to compare, in format from _mi_pack_key()
            This is normally key specified by user
    b       Second key to compare.  This is always from a row
    key_length  Length of key to compare.  This can be shorter than
                a to just compare sub keys
    next_flag   How keys should be compared
              If bit SEARCH_FIND is not set the keys includes the row
              position and this should also be compared
    diff_pos    OUT Number of first keypart where values differ, counting
                from one.
    diff_pos[1] OUT  (b + diff_pos[1]) points to first value in tuple b
                      that is different from corresponding value in tuple a.

  EXAMPLES
   Example1: if the function is called for tuples
     ('aaa','bbb') and ('eee','fff'), then
     diff_pos[0] = 1 (as 'aaa' != 'eee')
     diff_pos[1] = 0 (offset from beggining of tuple b to 'eee' keypart).

   Example2: if the index function is called for tuples
     ('aaa','bbb') and ('aaa','fff'),
     diff_pos[0] = 2 (as 'aaa' != 'eee')
     diff_pos[1] = 3 (offset from beggining of tuple b to 'fff' keypart,
                      here we assume that first key part is CHAR(3) NOT NULL)

  NOTES
    Number-keys can't be splited

  RETURN VALUES
    <0	If a < b
    0	If a == b
    >0	If a > b
*/

#define FCMP(A,B) ((int) (A) - (int) (B))

int ha_key_cmp(register HA_KEYSEG *keyseg, register unsigned char *a,
	       register unsigned char *b, uint32_t key_length, uint32_t nextflag,
	       uint32_t *diff_pos)
{
  int flag;
  int16_t s_1,s_2;
  int32_t l_1,l_2;
  uint32_t u_1,u_2;
  float f_1,f_2;
  double d_1,d_2;
  uint32_t next_key_length;
  unsigned char *orig_b= b;

  *diff_pos=0;
  for ( ; (int) key_length >0 ; key_length=next_key_length, keyseg++)
  {
    unsigned char *end;
    uint32_t piks=! (keyseg->flag & HA_NO_SORT);
    (*diff_pos)++;
    diff_pos[1]= (uint)(b - orig_b);

    /* Handle NULL part */
    if (keyseg->null_bit)
    {
      key_length--;
      if (*a != *b && piks)
      {
        flag = (int) *a - (int) *b;
        return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
      }
      b++;
      if (!*a++)                                /* If key was NULL */
      {
        if (nextflag == (SEARCH_FIND | SEARCH_UPDATE))
          nextflag=SEARCH_SAME;                 /* Allow duplicate keys */
        else if (nextflag & SEARCH_NULL_ARE_NOT_EQUAL)
        {
           /*
            This is only used from mi_check() to calculate cardinality.
             It can't be used when searching for a key as this would cause
            compare of (a,b) and (b,a) to return the same value.
            */
             return -1;
        }
        next_key_length=key_length;
        continue;                               /* To next key part */
      }
    }
    end= a+ cmin(keyseg->length,key_length);
    next_key_length=key_length-keyseg->length;

    switch ((enum ha_base_keytype) keyseg->type) {
    case HA_KEYTYPE_TEXT:                       /* Ascii; Key is converted */
      if (keyseg->flag & HA_SPACE_PACK)
      {
        int a_length,b_length,pack_length;
        get_key_length(a_length,a);
        get_key_pack_length(b_length,pack_length,b);
        next_key_length=key_length-b_length-pack_length;

        if (piks &&
            (flag=ha_compare_text(keyseg->charset,a,a_length,b,b_length,
				  (bool) ((nextflag & SEARCH_PREFIX) &&
					     next_key_length <= 0),
				  (bool)!(nextflag & SEARCH_PREFIX))))
          return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
        a+=a_length;
        b+=b_length;
        break;
      }
      else
      {
	uint32_t length=(uint) (end-a), a_length=length, b_length=length;
        if (piks &&
            (flag= ha_compare_text(keyseg->charset, a, a_length, b, b_length,
				   (bool) ((nextflag & SEARCH_PREFIX) &&
					      next_key_length <= 0),
				   (bool)!(nextflag & SEARCH_PREFIX))))
          return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
        a=end;
        b+=length;
      }
      break;
    case HA_KEYTYPE_BINARY:
    case HA_KEYTYPE_BIT:
      if (keyseg->flag & HA_SPACE_PACK)
      {
        int a_length,b_length,pack_length;
        get_key_length(a_length,a);
        get_key_pack_length(b_length,pack_length,b);
        next_key_length=key_length-b_length-pack_length;

        if (piks &&
	    (flag=compare_bin(a,a_length,b,b_length,
                              (bool) ((nextflag & SEARCH_PREFIX) &&
                                         next_key_length <= 0),1)))
          return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
        a+=a_length;
        b+=b_length;
        break;
      }
      else
      {
        uint32_t length=keyseg->length;
        if (piks &&
	    (flag=compare_bin(a,length,b,length,
                              (bool) ((nextflag & SEARCH_PREFIX) &&
                                         next_key_length <= 0),0)))
          return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
        a+=length;
        b+=length;
      }
      break;
    case HA_KEYTYPE_VARTEXT1:
    case HA_KEYTYPE_VARTEXT2:
      {
        int a_length,b_length,pack_length;
        get_key_length(a_length,a);
        get_key_pack_length(b_length,pack_length,b);
        next_key_length=key_length-b_length-pack_length;

        if (piks &&
	    (flag= ha_compare_text(keyseg->charset,a,a_length,b,b_length,
                                   (bool) ((nextflag & SEARCH_PREFIX) &&
                                              next_key_length <= 0),
				   (bool) ((nextflag & (SEARCH_FIND |
							   SEARCH_UPDATE)) ==
					      SEARCH_FIND &&
                                              ! (keyseg->flag &
                                                 HA_END_SPACE_ARE_EQUAL)))))
          return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
        a+= a_length;
        b+= b_length;
        break;
      }
    case HA_KEYTYPE_VARBINARY1:
    case HA_KEYTYPE_VARBINARY2:
      {
        int a_length,b_length,pack_length;
        get_key_length(a_length,a);
        get_key_pack_length(b_length,pack_length,b);
        next_key_length=key_length-b_length-pack_length;

        if (piks &&
	    (flag=compare_bin(a,a_length,b,b_length,
                              (bool) ((nextflag & SEARCH_PREFIX) &&
                                         next_key_length <= 0), 0)))
          return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
        a+=a_length;
        b+=b_length;
        break;
      }
    case HA_KEYTYPE_INT8:
    {
      int i_1= (int) *((signed char*) a);
      int i_2= (int) *((signed char*) b);
      if (piks && (flag = CMP_NUM(i_1,i_2)))
        return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
      a= end;
      b++;
      break;
    }
    case HA_KEYTYPE_SHORT_INT:
      s_1= mi_sint2korr(a);
      s_2= mi_sint2korr(b);
      if (piks && (flag = CMP_NUM(s_1,s_2)))
        return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
      a=  end;
      b+= 2; /* sizeof(short int); */
      break;
    case HA_KEYTYPE_USHORT_INT:
      {
        uint16_t us_1,us_2;
        us_1= mi_sint2korr(a);
        us_2= mi_sint2korr(b);
        if (piks && (flag = CMP_NUM(us_1,us_2)))
          return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
        a=  end;
        b+=2; /* sizeof(short int); */
        break;
      }
    case HA_KEYTYPE_LONG_INT:
      l_1= mi_sint4korr(a);
      l_2= mi_sint4korr(b);
      if (piks && (flag = CMP_NUM(l_1,l_2)))
        return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
      a=  end;
      b+= 4; /* sizeof(long int); */
      break;
    case HA_KEYTYPE_ULONG_INT:
      u_1= mi_sint4korr(a);
      u_2= mi_sint4korr(b);
      if (piks && (flag = CMP_NUM(u_1,u_2)))
        return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
      a=  end;
      b+= 4; /* sizeof(long int); */
      break;
    case HA_KEYTYPE_INT24:
      l_1=mi_sint3korr(a);
      l_2=mi_sint3korr(b);
      if (piks && (flag = CMP_NUM(l_1,l_2)))
        return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
      a=  end;
      b+= 3;
      break;
    case HA_KEYTYPE_UINT24:
      l_1=mi_uint3korr(a);
      l_2=mi_uint3korr(b);
      if (piks && (flag = CMP_NUM(l_1,l_2)))
        return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
      a=  end;
      b+= 3;
      break;
    case HA_KEYTYPE_FLOAT:
      mi_float4get(f_1,a);
      mi_float4get(f_2,b);
      /*
        The following may give a compiler warning about floating point
        comparison not being safe, but this is ok in this context as
        we are bascily doing sorting
      */
      if (piks && (flag = CMP_NUM(f_1,f_2)))
        return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
      a=  end;
      b+= 4; /* sizeof(float); */
      break;
    case HA_KEYTYPE_DOUBLE:
      mi_float8get(d_1,a);
      mi_float8get(d_2,b);
      /*
        The following may give a compiler warning about floating point
        comparison not being safe, but this is ok in this context as
        we are bascily doing sorting
      */
      if (piks && (flag = CMP_NUM(d_1,d_2)))
        return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
      a=  end;
      b+= 8;  /* sizeof(double); */
      break;
    case HA_KEYTYPE_NUM:                                /* Numeric key */
    {
      int swap_flag= 0;
      int alength,blength;

      if (keyseg->flag & HA_REVERSE_SORT)
      {
        swap_variables(unsigned char*, a, b);
        swap_flag=1;                            /* Remember swap of a & b */
        end= a+ (int) (end-b);
      }
      if (keyseg->flag & HA_SPACE_PACK)
      {
        alength= *a++; blength= *b++;
        end=a+alength;
        next_key_length=key_length-blength-1;
      }
      else
      {
        alength= (int) (end-a);
        blength=keyseg->length;
        /* remove pre space from keys */
        for ( ; alength && *a == ' ' ; a++, alength--) ;
        for ( ; blength && *b == ' ' ; b++, blength--) ;
      }
      if (piks)
      {
	if (*a == '-')
	{
	  if (*b != '-')
	    return -1;
	  a++; b++;
	  swap_variables(unsigned char*, a, b);
	  swap_variables(int, alength, blength);
	  swap_flag=1-swap_flag;
	  alength--; blength--;
	  end=a+alength;
	}
	else if (*b == '-')
	  return 1;
	while (alength && (*a == '+' || *a == '0'))
	{
	  a++; alength--;
	}
	while (blength && (*b == '+' || *b == '0'))
	{
	  b++; blength--;
	}
	if (alength != blength)
	  return (alength < blength) ? -1 : 1;
	while (a < end)
	  if (*a++ !=  *b++)
	    return ((int) a[-1] - (int) b[-1]);
      }
      else
      {
        b+=(end-a);
        a=end;
      }

      if (swap_flag)                            /* Restore pointers */
        swap_variables(unsigned char*, a, b);
      break;
    }
    case HA_KEYTYPE_LONGLONG:
    {
      int64_t ll_a,ll_b;
      ll_a= mi_sint8korr(a);
      ll_b= mi_sint8korr(b);
      if (piks && (flag = CMP_NUM(ll_a,ll_b)))
        return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
      a=  end;
      b+= 8;
      break;
    }
    case HA_KEYTYPE_ULONGLONG:
    {
      uint64_t ll_a,ll_b;
      ll_a= mi_uint8korr(a);
      ll_b= mi_uint8korr(b);
      if (piks && (flag = CMP_NUM(ll_a,ll_b)))
        return ((keyseg->flag & HA_REVERSE_SORT) ? -flag : flag);
      a=  end;
      b+= 8;
      break;
    }
    case HA_KEYTYPE_END:                        /* Ready */
      goto end;                                 /* diff_pos is incremented */
    }
  }
  (*diff_pos)++;
end:
  if (!(nextflag & SEARCH_FIND))
  {
    uint32_t i;
    if (nextflag & (SEARCH_NO_FIND | SEARCH_LAST)) /* Find record after key */
      return (nextflag & (SEARCH_BIGGER | SEARCH_LAST)) ? -1 : 1;
    flag=0;
    for (i=keyseg->length ; i-- > 0 ; )
    {
      if (*a++ != *b++)
      {
        flag= FCMP(a[-1],b[-1]);
        break;
      }
    }
    if (nextflag & SEARCH_SAME)
      return (flag);                            /* read same */
    if (nextflag & SEARCH_BIGGER)
      return (flag <= 0 ? -1 : 1);              /* read next */
    return (flag < 0 ? -1 : 1);                 /* read previous */
  }
  return 0;
} /* ha_key_cmp */


/*
  Find the first NULL value in index-suffix values tuple

  SYNOPSIS
    ha_find_null()
      keyseg     Array of keyparts for key suffix
      a          Key suffix value tuple

  DESCRIPTION
    Find the first NULL value in index-suffix values tuple.

  TODO
    Consider optimizing this function or its use so we don't search for
    NULL values in completely NOT NULL index suffixes.

  RETURN
    First key part that has NULL as value in values tuple, or the last key
    part (with keyseg->type==HA_TYPE_END) if values tuple doesn't contain
    NULLs.
*/

HA_KEYSEG *ha_find_null(HA_KEYSEG *keyseg, unsigned char *a)
{
  for (; (enum ha_base_keytype) keyseg->type != HA_KEYTYPE_END; keyseg++)
  {
    unsigned char *end;
    if (keyseg->null_bit)
    {
      if (!*a++)
        return keyseg;
    }
    end= a+ keyseg->length;

    switch ((enum ha_base_keytype) keyseg->type) {
    case HA_KEYTYPE_TEXT:
    case HA_KEYTYPE_BINARY:
    case HA_KEYTYPE_BIT:
      if (keyseg->flag & HA_SPACE_PACK)
      {
        int a_length;
        get_key_length(a_length, a);
        a += a_length;
        break;
      }
      else
        a= end;
      break;
    case HA_KEYTYPE_VARTEXT1:
    case HA_KEYTYPE_VARTEXT2:
    case HA_KEYTYPE_VARBINARY1:
    case HA_KEYTYPE_VARBINARY2:
      {
        int a_length;
        get_key_length(a_length, a);
        a+= a_length;
        break;
      }
    case HA_KEYTYPE_NUM:
      if (keyseg->flag & HA_SPACE_PACK)
      {
        int alength= *a++;
        end= a+alength;
      }
      a= end;
      break;
    case HA_KEYTYPE_INT8:
    case HA_KEYTYPE_SHORT_INT:
    case HA_KEYTYPE_USHORT_INT:
    case HA_KEYTYPE_LONG_INT:
    case HA_KEYTYPE_ULONG_INT:
    case HA_KEYTYPE_INT24:
    case HA_KEYTYPE_UINT24:
    case HA_KEYTYPE_LONGLONG:
    case HA_KEYTYPE_ULONGLONG:
    case HA_KEYTYPE_FLOAT:
    case HA_KEYTYPE_DOUBLE:
      a= end;
      break;
    case HA_KEYTYPE_END:                        /* purecov: inspected */
      /* keep compiler happy */
      assert(0);
      break;
    }
  }
  return keyseg;
}




